package yz.mr.ComBine;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Combine {
    public static void main(String[] args) throws Exception{

        //1.创建Job操作对象
        //2.创建配置类对象
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        //3.配置job
        job.setJobName("Combine");
        job.setJarByClass(Combine.class);

        //4.设置Mapper以及输出类型
        job.setMapperClass(CombineMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5.设置Reduce以及输出对象
        job.setReducerClass(CombineReducer.class);

        //6.设置最终地输出数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        /**
         * combine的作用：
         *      将Reduce端的处理逻辑提前到Mapper端执行
         *
         *
         * 优点：
         *      减少了Mapper端的输出的数据量
         *      减少了IO压力
         *      提高了计算速度
         */
        //当Reduce端的处理逻辑可以提前时，才可以使用Combine作为优化，并不是所有的都可以
        job.setCombinerClass(MyCombine.class);


        //本地执行路径
        FileInputFormat.addInputPath(job,new Path("src/main/dataFile/apiwolds.txt"));
        FileOutputFormat.setOutputPath(job,new Path("output/Combine"));

        //8.提交执行当前JOB
        job.waitForCompletion(true);

    }
}
